package com.openess.bigsearch.engine.index;

import java.io.IOException;

import com.openess.bigsearch.engine.BPlusTree.BPlusTree;
import com.openess.bigsearch.engine.BPlusTree.BPlusTree.BTree_node;
import com.openess.bigsearch.engine.BPlusTree.StringBKCompare;
import com.openess.bigsearch.engine.BTree.util.Arrays;
import com.openess.bigsearch.engine.BTree.util.Arrays.Sort;
import com.openess.bigsearch.engine.Text.Text;
import com.openess.bigsearch.engine.analyzer.Analyzer;
import com.openess.bigsearch.engine.document.Document;
import com.openess.bigsearch.engine.document.Field;
import com.openess.bigsearch.engine.index.merge.Merge;
import com.openess.bigsearch.engine.utils.HashMap;
import com.openess.bigsearch.engine.utils.LinkedList;
import com.openess.bigsearch.engine.utils.Map;
import com.openess.bigsearch.engine.utils.Map.Entry;

public class IndexWriter {
	/**
	 * 依据BPlusTree设置分词的长度，对超过长度的进行截取
	 */
	public static final int TERMLENGTH = BTree_node.STRSIZE;
	public static final int docMax = 8000;
	/**
	 * B+树，利用B+树来存储数据，写成索引
	 */
	private BPlusTree tree;

	/**
	 * 散列表，利用散列表的插入/查询复杂度为O(1)的特点，并结合set集合特性：key值唯一。快速的将一个个单独的token整合在一起，比起原先设计的
	 * ，先排序再合并，效率要高。此处选择链表而不是线性表的主要原因：1.链表比线性表节省空间。2.在合并的时候线性表还需要重新创建数组并导入数据，
	 * 但链表可以直接合并，合并时的时间复杂度低。
	 */
	private HashMap<String, LinkedList<FrqPos>> map;
	/**
	 * Nrm对象，在这里用于写入磁盘，而不是读
	 */
	private Nrm nrm;

	/**
	 * 专门用来处理范围表，比如写入，加载，整合
	 */
	private Extent ext;
	/**
	 * 用于保存域文本
	 */
	private Text text;
	/**
	 * 用于统计上传了几篇文档
	 */
	private int docNum;
	/**
	 * Segment对象，用于写入关于合并的信息，属于整个索引的元数据
	 */
	private Segment seg;
	/**
	 * 分词解析器
	 */
	private Analyzer an;
	/**
	 * 索引中总文档长度
	 */
	private int totalLength;
	/**
	 * 用于保存(docID, Length)对的散列表，主要为了nrm文件写入文档长度
	 */
	private HashMap<Long, Integer> nrmap;
	/**
	 * 索引的根目录
	 */
	private String indexPath;
	/**
	 * 记录下一个待写入的版本号，用于索引合并中，区分每个临时索引
	 */
	private int version;
	private int docCount;

	/**
	 * 构造器，创建B+树，创建散列表，创建些范围表的输出流
	 * 
	 * @param indexPath
	 *            往哪里写索引
	 * @param an
	 *            采用什么解析器来对文章进行分词
	 * @throws IOException
	 */
	public IndexWriter(String indexPath, Analyzer an) throws IOException {
		this.an = an;
		this.indexPath = indexPath;
		map = new HashMap<String, LinkedList<FrqPos>>(500);
		seg = new Segment(indexPath);
		version = seg.getVersion();
		ext = new Extent(indexPath, "w", version);
		text = new Text(indexPath, "w", version);
		nrm = new Nrm(indexPath, "w", version);
		tree = new BPlusTree(indexPath, new StringBKCompare(), version);
		nrmap = new HashMap<Long, Integer>(200);
	}

	/**
	 * 写索引，采用B+树结构
	 * 
	 * @throws IOException
	 */
	public void write() throws IOException {
		for (Entry<String, LinkedList<FrqPos>> en : map.entrySet()) {
			// 将倒排表按照DocID进行排序
			FrqPos[] array = sort(en.getValue());
			tree.insert(en.getKey(), array);
		}
		// 计算向量模长
		docNorm();
		// 将文档总数和总传给Segment对象，并准备写入磁盘
		nrm.setDocNum(docNum);
		nrm.setTotalLength(totalLength);
		// 写入范围表
		ext.writeExtent();
		// 写入所有的域文本
		text.write();
		// 写入nmr表
		nrm.write();
		close();
	}

	/**
	 * 内部其实还是调用write()方法和merge()方法，但write()是用户调用的，commit()才是用户调用的接口
	 * 
	 * @throws IOException
	 */
	public void commit() throws IOException {
		write();
		clear();
		version++;
		Merge.merge(indexPath, version);
		// 最后合并完再写入版本
		seg.write();
		seg.close();
	}

	/**
	 * 对链表进行排序
	 * 
	 * @param token
	 * @return
	 */
	private FrqPos[] sort(LinkedList<FrqPos> list) {
		FrqPos[] value = list.toArray(new FrqPos[0]);
		// 利用堆排序
		Arrays.sort(value, Sort.HEAP);
		return value;
	}

	/**
	 * 添加一篇文档，并对文档进行分词，规整，最终组成倒排表。是appDoc的优化版，去除了大量的HashMap，改用数组，去除以及合并都改用排序方式实现
	 * 
	 * @param docID
	 *            文档ID
	 * @param input
	 *            待进行分词的文档具体内容
	 * @throws IOException
	 */
	public void addDocOP(Document doc) throws IOException {
		// 内存满了，就写出索引
		if (docCount++ > docMax) {
			write();
			version++;
			reset();
			docCount = 0;
		}
		// 每调用一次addDoc方法，文档数就加1
		docNum++;
		long id = doc.getID();
		// 统计索引总文档长度
		totalLength += doc.getLength();
		// 装入散列表，准备写入nrm文件
		nrmap.put(id, doc.getLength());
		Field[] fields = doc.getFields();
		// 每个域的起始位置
		int startOffset = 0;
		// 用于装载待合并的倒排表
		LinkedList<Data> list = new LinkedList<Data>();
		// 保存这篇文档的所有关键域的长度
		Integer[] fL = new Integer[Field.MFSIZE];
		for (Field field : fields) {
			// 判断这个域是否为关键域
			int fieldIndex = Field.indexField(field.getFieldName());
			if (fieldIndex >= 0)
				fL[fieldIndex] = field.size();
			// 对文档内容进行分词解析，过滤
			Token<String, Integer>[] tokens = an.parse(startOffset,
					field.getContent());
			if (tokens == null)
				continue;
			LinkedList<Data> temp = mergePos(tokens, id, fieldIndex);
			list.addAll(temp);
			// 添加域的范围表
			ext.addExtent(field.getFieldName(), id, startOffset, startOffset
					+ field.size());
			// 下一个域的起始偏移量
			startOffset += field.size();
		}
		// 单独一篇文档的最终倒排表
		LinkedList<Data> dlist = merge(list);
		// 将单独一篇文档的倒排表放到总倒排表中
		for (Data d : dlist) {
			String term = d.term;
			FrqPos fp = d.toFP();
			LinkedList<FrqPos> value = map.get(term);
			if (value == null) {
				value = new LinkedList<FrqPos>();
				value.add(fp);
				map.put(term, value);
			} else
				value.add(fp);
		}
		// 将本文档的关键域长度，添加到Extent对象中
		ext.addMainExt(id, fL);
	}

	/**
	 * 将(term, pos)合并成(term, pos[])，也统计ffrqList
	 * 
	 * @param token
	 * @param id
	 * @param fieldIndex
	 */
	private LinkedList<Data> mergePos(Token<String, Integer>[] tokens, long id,
			int fieldIndex) {
		LinkedList<Data> list = new LinkedList<Data>();
		Arrays.sort(tokens, Sort.QUICK);
		Token<String, Integer> first = tokens[0];
		Data d = new Data(first.getKey(), id);
		d.addPos(first.getValue());
		list.add(d);
		Token<String, Integer> second = null;
		for (int i = 1; i < tokens.length; i++) {
			second = tokens[i];
			if (d.term.equals(second.getKey()))
				d.addPos(second.getValue());
			else {
				d.setFfrq(fieldIndex, d.posSize());
				d = new Data(second.getKey(), id);
				d.addPos(second.getValue());
				list.add(d);
			}
		}
		return list;
	}

	/**
	 * 最终的合并，生成完整的倒排表
	 * 
	 * @param list
	 * @return
	 */
	private LinkedList<Data> merge(LinkedList<Data> list) {
		LinkedList<Data> result = new LinkedList<Data>();
		Data[] array = list.toArray(list.get(0));
		Arrays.sort(array, Sort.QUICK);
		Data d = array[0];
		Data next = null;
		result.add(d);
		for (int i = 1; i < array.length; i++) {
			next = array[i];
			if (d.equals(next)) {
				d.addPos(next.posList);
				for (int j = 0; j < Field.MFSIZE; j++)
					d.setFfrq(j, next.ffrqList[j]);
			} else {
				d = next;
				result.add(d);
			}
		}
		return result;
	}

	/**
	 * 添加一篇文档，并对文档进行分词，规整，最终组成Token<String,
	 * Integer[]>的形式，如果多次添加同一个docID，那么会把之前的数据覆盖掉
	 * 
	 * @param docID
	 *            文档ID
	 * @param input
	 *            待进行分词的文档具体内容
	 * @throws IOException
	 */
	public void addDoc(Document doc) throws IOException {
		// 内存满了，就写出索引
		if (docCount++ > docMax) {
			write();
			version++;
			reset();
			docCount = 0;
		}
		// 每调用一次addDoc方法，文档数就加1
		docNum++;
		long id = doc.getID();
		// 统计索引总文档长度
		totalLength += doc.getLength();
		// 装入散列表，准备写入nrm文件
		nrmap.put(id, doc.getLength());
		Field[] fields = doc.getFields();
		// 每个域的起始位置
		int startOffset = 0;
		// 用于统计一篇文档中位置的集合，每次调用addDoc都要清空集合，因为词频和位置信息都是只针对一篇文档
		HashMap<String, LinkedList<Integer>> posSet = new HashMap<String, LinkedList<Integer>>();
		// 保存有(term, ffrqList)
		HashMap<String, Integer[]> posSetFL = new HashMap<String, Integer[]>();
		// 保存这篇文档的所有关键域的长度
		Integer[] fL = new Integer[Field.MFSIZE];
		for (Field field : fields) {
			if (field.isStored())
				text.put(id, field.getFieldName(), field.getContent());
			// 判断这个域是否为关键域
			int fieldIndex = Field.indexField(field.getFieldName());
			// 功能和posSet基本一致，区别在于：posSet统计的是分词在整篇文档的所有位置，而posSetF统计的是分词在一篇文档的关键
			// 域中的词频，主要用于BM25F模型
			HashMap<String, Integer> posSetF = new HashMap<String, Integer>();
			// 对文档内容进行分词解析，过滤
			Token<String, Integer>[] tokens = an.parse(startOffset,
					field.getContent());
			// 添加域的范围表
			ext.addExtent(field.getFieldName(), id, startOffset, startOffset
					+ field.size());
			if (tokens == null)
				continue;
			// 组合(term, pos)对
			for (Token<String, Integer> token : tokens) {
				String term = token.getKey();
				// 截取超过指定长度的分词
				if (term.length() > TERMLENGTH)
					term = term.substring(0, TERMLENGTH);
				int position = token.getValue();
				// 将相同term的(term, pos)对，利用HashMap整合在一起，组成(term, pos[])的形式
				LinkedList<Integer> value = posSet.get(term);
				if (value == null) {
					value = new LinkedList<Integer>();
					value.add(position);
					posSet.put(term, value);
				} else
					value.add(position);
				if (fieldIndex >= 0) {
					// 统计域词频
					Integer count = posSetF.get(term);
					if (count == null)
						count = 1;
					else
						count += 1;
					posSetF.put(term, count);
				}
			}
			if (fieldIndex >= 0) {
				fL[fieldIndex] = field.size();
				// 循环将全部关键域的数据装填在域词频数组中
				for (Entry<String, Integer> e : posSetF.entrySet()) {
					String term = e.getKey();
					Integer ffrq = e.getValue();
					// 关键域的词频数组
					Integer[] ffrqList = posSetFL.get(term);
					if (ffrqList == null) {
						ffrqList = new Integer[Field.MFSIZE];
						ffrqList[fieldIndex] = ffrq;
						posSetFL.put(term, ffrqList);
					} else
						ffrqList[fieldIndex] = ffrq;
				}
			}
			// 下一个域的起始偏移量
			startOffset += field.size();
		}
		// 将本文档的关键域长度，添加到Extent对象中
		ext.addMainExt(id, fL);
		// 将term，id，frq，frqList，pos数据全部整合
		process(id, posSet, posSetFL);
	}

	/**
	 * 将文档ID与(分词，位置[])，(分词，域词频[])数据合并，合并为(分词：[ID1:ffrqList (pos1 ,
	 * pos2)]，[ID1:ffrqList (pos1,pos2)])的形式
	 * 。每个ID1:ffrqList(pos1,pos2)包装成对象FrqPos，组合成Token<term,
	 * LinkedList<FrqPos>>放入set集合中，在set中将同属一个term的FrqPos链表整合
	 * 
	 * @param id
	 *            文档ID
	 * @param posSet
	 *            保存有(分词，位置)的集合
	 * @param posSetFL
	 *            保存有(term, ffrqList)
	 */
	private void process(long id, HashMap<String, LinkedList<Integer>> posSet,
			HashMap<String, Integer[]> posSetFL) {
		// 拿到保存有term,pos的数据集合，取出数据，重新整合
		for (Entry<String, LinkedList<Integer>> e : posSet.entrySet()) {
			String term = e.getKey();
			// posSet保存的是所有域的分词，但posSetFL仅保存了关键域中的分词，因此倒排表中有的term其关键域数据全为空
			Integer[] ffrqList = posSetFL.get(term);
			Integer[] pos = e.getValue().toArray(new Integer[0]);
			if (ffrqList == null)
				ffrqList = new Integer[Field.MFSIZE];
			// 追加value
			LinkedList<FrqPos> value = map.get(term);
			if (value == null) {
				value = new LinkedList<FrqPos>();
				value.add(new FrqPos(id, ffrqList, pos));
				map.put(term, value);
			} else
				value.add(new FrqPos(id, ffrqList, pos));
		}
	}

	/**
	 * 计算每个文档的向量模长度，利用term-at-a-time查询处理算法，并将最终数据写入nrm文件中
	 * 
	 */
	private void docNorm() {
		HashMap<Long, Double> docSet = new HashMap<Long, Double>();
		for (Map.Entry<String, LinkedList<FrqPos>> e : map.entrySet()) {
			LinkedList<FrqPos> list = e.getValue();
			// 词项出现过的文档个数
			int n = list.size();
			// 逆文档频率，只有一个变量：词项
			double idf = Arrays.IDF(n, docNum);
			for (FrqPos fp : list) {
				long id = fp.getId();
				double tf = Arrays.TF(fp.getFrq());
				// 词项量特征的平方
				double d2 = Math.pow(tf * idf, 2);
				docSet.put(id, d2);
			}
		}
		for (Map.Entry<Long, Double> e : docSet.entrySet()) {
			long docID = e.getKey();
			double norm = Math.sqrt(e.getValue());
			int length = nrmap.get(docID);
			nrm.add(docID, norm, length);
		}
	}

	/**
	 * 拿到保存的分词数据，待写入到B+树中的临时集合
	 * 
	 * @return
	 */
	public HashMap<String, LinkedList<FrqPos>> getSet() {
		return this.map;
	}

	/**
	 * 关闭所有流资源
	 * 
	 * @throws IOException
	 */
	public void close() throws IOException {
		tree.close();
	}

	/**
	 * 重建所有资源，用于写第二份索引(重建前先关闭所有流资源)
	 * 
	 * @throws IOException
	 */
	public void reset() throws IOException {
		tree = new BPlusTree(indexPath, new StringBKCompare(), version);
		map = new HashMap<String, LinkedList<FrqPos>>(500);
		ext = new Extent(indexPath, "w", version);
		nrm = new Nrm(indexPath, "w", version);
		nrmap = new HashMap<Long, Integer>(200);
	}

	/**
	 * 清空拥有大数据量的变量
	 */
	public void clear() {
		tree = null;
		map = null;
		ext = null;
		text = null;
		nrm = null;
		nrmap = null;
	}

	private class Data implements Comparable<Data> {
		private String term;
		private long id;
		private Integer[] ffrqList;
		/**
		 * pos不用排序，因为本身就是按顺序加入集合的
		 */
		private LinkedList<Integer> posList;

		public Data(String term, long id) {
			this.term = term;
			this.id = id;
			ffrqList = new Integer[Field.MFSIZE];
			posList = new LinkedList<Integer>();
		}

		/**
		 * 将当前Data转换为FrqPos
		 * 
		 * @return
		 */
		public FrqPos toFP() {
			Integer[] poslist = posList.toArray(1);
			return new FrqPos(id, ffrqList, poslist);
		}

		/**
		 * 给关键域添加域词频，如果index<0或ffrq == 0直接返回
		 * 
		 * @param index
		 *            指定的关键域
		 * @param ffrq
		 */
		public void setFfrq(int index, Integer ffrq) {
			if (index < 0 || ffrq == null)
				return;
			ffrqList[index] = ffrq;
		}

		public void addPos(int pos) {
			posList.add(pos);
		}

		public void addPos(LinkedList<Integer> posList) {
			posList.addAll(posList);
		}

		public int posSize() {
			return posList.size();
		}

		@Override
		public boolean equals(Object ob) {
			return term.equals(((Data) ob).term);
		}

		@Override
		public int compareTo(Data o) {
			return term.compareTo(o.term);
		}
	}
}
